package com.zhangsr.zk.client;

import com.zhangsr.zk.client.exception.ZkException;
import org.apache.log4j.Logger;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Author: zhangsr
 */
public class ZkConnection implements IZkConnection {

    private final static Logger log = Logger.getLogger(ZkConnection.class);

    private Lock zkLock = new ReentrantLock();

    /**
     * zookeeper instance
     */
    private ZooKeeper zk = null;

    private String zkServers;
    private int sessionTimeout;

    public ZkConnection(String zkServers, int sessionTimeout) {
        this.zkServers = zkServers;
        this.sessionTimeout = sessionTimeout;
    }

    @Override
    public void connect(Watcher watcher) throws IOException {
        zkLock.lock();
        try {
            if (zk != null) {
                throw new IllegalArgumentException("zk client has already been started.");
            }
            try {
                log.debug("Creating new Zookeeper instance to connect to " + zkServers + ".");
                zk = new ZooKeeper(zkServers, sessionTimeout, watcher);
            } catch (IOException e) {
                throw new ZkException("Unable to connect to " + zkServers, e);
            }
        } finally {
            zkLock.unlock();
        }
        log.info("connected zk! conf = " + zkServers);
    }

    @Override
    public void close() throws InterruptedException {
        zkLock.lock();
        try {
            if (zk != null) {
                log.debug("Closing Zookeeper connected to " + zkServers);
                zk.close();
                zk = null;
            }
        } finally {
            zkLock.unlock();
        }
        log.info("closed zk!");
    }

    @Override
    public boolean exists(String path, boolean watch) throws KeeperException, InterruptedException {
        return zk.exists(path, watch) != null;
    }

    @Override
    public String create(String path, byte[] data, CreateMode mode) throws KeeperException, InterruptedException {
        return zk.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, mode);
    }

    @Override
    public byte[] readData(String path, Stat stat, boolean watch) throws KeeperException, InterruptedException {
        return zk.getData(path, watch, stat);
    }
}
